package org.dbstar.scheduler.control.db.job;

import java.io.File;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Map;

import net.sf.json.JSONObject;
import net.sf.json.JSONSerializer;

import org.dbstar.scheduler.beans.CmdJob;
import org.dbstar.scheduler.beans.CmdJobConf;
import org.dbstar.scheduler.beans.CompoundJob;
import org.dbstar.scheduler.beans.CompoundJobConf;
import org.dbstar.scheduler.beans.Job;
import org.dbstar.scheduler.control.db.MapJsonConfig;
import org.dbstar.scheduler.control.db.ObjectRowMapper;
import org.dbstar.scheduler.identifier.DefaultIdentifier;
import org.dbstar.scheduler.job.JobStatus;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.object.SqlQuery;


/**
 * Job的Query对象
 * 
 * @author 代波
 * @since 1.0.0
 * @version 1.0.0
 * @date 2012-4-24
 */
public class JobQuery extends SqlQuery<Job> {
	private final JobRowMapper jobRowMapper;

	public JobQuery(JdbcTemplate jdbcTemplate, File resultRoot) {
		this(jdbcTemplate, null, resultRoot);
	}

	public JobQuery(JdbcTemplate jdbcTemplate, String condition, File resultRoot) {
		super();
		setJdbcTemplate(jdbcTemplate);
		setSql("SELECT * FROM SCHD_JOB" + (condition == null ? "" : (" " + condition)));
		jobRowMapper = new JobRowMapper(resultRoot);
	}

	@Override
	protected final RowMapper<Job> newRowMapper(Object[] parameters, @SuppressWarnings("rawtypes") Map context) {
		return jobRowMapper;
	}

	public static class JobRowMapper extends ObjectRowMapper<Job> {
		private final File resultRoot;

		public JobRowMapper(File resultRoot) {
			super();
			this.resultRoot = resultRoot;
		}

		@Override
		public Job mapRow(ResultSet rs, int rowNum) throws SQLException {
			Job job = buildJob(rs, rs.getString("TYPE"));
			job.setIdentifier(DefaultIdentifier.valueOf(rs.getString("IDENTIFIER")));
			job.setJobConfIdentifier(DefaultIdentifier.valueOf(rs.getString("CONF_IDENTIFIER")));
			job.setStatus(JobStatus.valueOf(rs.getString("STATUS")));
			job.setResource(rs.getString("RESOURCE"));
			job.setContent(rs.getString("CONTENT"));
			job.setParameters(readParameters(rs.getString("PARAMETER_DATA")));
			job.setSubmitTime(rs.getLong("SUBMIT_TIME"));
			job.setLaunchTime(getLongObject(rs, "LAUNCH_TIME"));
			job.setFinishTime(getLongObject(rs, "FINISH_TIME"));
			job.setReady(rs.getBoolean("READY"));
			job.setResultString(rs.getString("RESULT"));
			job.setRetry(rs.getInt("RETRY"));
			job.setMaxRetry(rs.getInt("MAX_RETRY"));
			job.setSubmitter(rs.getString("SUBMITTER"));
			return job;
		}

		@SuppressWarnings("unchecked")
		private Map<String, String> readParameters(String str) throws SQLException {
			if (str == null) {
				return null;
			} else {
				JSONObject obj = JSONObject.fromObject(str);
				return (Map<String, String>) JSONSerializer.toJava(obj, new MapJsonConfig());
			}
		}

		private Job buildJob(ResultSet rs, String type) throws SQLException {
			if (CmdJobConf.JOB_TYPE.equals(type)) {
				CmdJob job = new CmdJob();
				job.getResult().setResultRoot(resultRoot);
				return job;
			} else if (CompoundJobConf.JOB_TYPE.equals(type)) {
				return new CompoundJob();
			} else {
				throw new SQLException("Unknown job type: " + type);
			}
		}
	}
}
